package mr

import (
	"log"
	"net"
	"net/http"
	"net/rpc"
	"os"
	"sync"
	"time"
)

// const value for taskstatus
const (
	AVAIL    = 0
	WORKING  = 1
	FINISHED = 2
)

//Master struct
type Master struct {
	// Your definitions here.
	nReduce       int
	inputfiles    []string
	tasksfinished int   //amount of finished tasks
	taskstate     []int //index: taskid, value: state(0: available, 1: working, 2: finished)
	ismap         bool
}

var taskmutex sync.Mutex

// Your code here -- RPC handlers for the worker to call.

//Example :
// an example RPC handler.
//
// the RPC argument and reply types are defined in rpc.go.
//
func (m *Master) Example(args *ExampleArgs, reply *ExampleReply) error {
	reply.Y = args.X + 1
	return nil
}

//Distributetask :
func (m *Master) Distributetask(args *JobDoneArgs, reply *AsktaskReply) error {
	taskmutex.Lock()
	reply.Taskid, reply.Filename = m.SheduleTask()
	if reply.Taskid == -1 {
		reply.Status = 2
	} else if m.ismap {
		reply.Status = 0
		go m.WatchTask(reply.Taskid, true)
	} else {
		reply.Status = 1
		go m.WatchTask(reply.Taskid, false)
	}
	taskmutex.Unlock()
	return nil
}

//SheduleTask :
func (m *Master) SheduleTask() (int, string) {
	for index, status := range m.taskstate {
		if status == AVAIL {
			m.taskstate[index] = WORKING
			if m.ismap {
				return index, m.inputfiles[index]
			}
			return index, ""
		}
	}
	return -1, ""
}

//JobDone :
func (m *Master) JobDone(args *JobDoneArgs, reply *AsktaskReply) error {
	taskmutex.Lock()
	m.tasksfinished++
	m.taskstate[args.Taskid] = FINISHED
	if m.tasksfinished >= len(m.inputfiles) { //map is done, start reducing
		if m.ismap {
			m.ismap = false
			m.tasksfinished = 0
			i := 0
			for i < m.nReduce {
				if i < len(m.taskstate) {
					m.taskstate[i] = AVAIL
				} else {
					m.taskstate = append(m.taskstate, AVAIL)
				}
				i++
			}
		}
	}
	taskmutex.Unlock()
	return nil
}

//WatchTask :
//start a thread that monitor weather the task is finished
func (m *Master) WatchTask(taskid int, ismap bool) {
	time.Sleep(time.Second * 10)
	taskmutex.Lock()
	if ismap == m.ismap && m.taskstate[taskid] != FINISHED { //the task didn't finish
		m.taskstate[taskid] = AVAIL
	}
	taskmutex.Unlock()
}

//server :
// start a thread that listens for RPCs from worker.go
//
func (m *Master) server() {
	rpc.Register(m)
	rpc.HandleHTTP()
	//l, e := net.Listen("tcp", ":1234")
	sockname := masterSock()
	os.Remove(sockname)
	l, e := net.Listen("unix", sockname)
	if e != nil {
		log.Fatal("listen error:", e)
	}
	go http.Serve(l, nil)
}

// Done :
// main/mrmaster.go calls Done() periodically to find out
// if the entire job has finished.
//
func (m *Master) Done() bool {
	ret := false

	// Your code here.
	taskmutex.Lock()
	if !m.ismap && m.tasksfinished >= m.nReduce {
		ret = true
	}
	taskmutex.Unlock()

	return ret
}

// MakeMaster :
// create a Master.
// main/mrmaster.go calls this function.
// nReduce is the number of reduce tasks to use.
//
func MakeMaster(files []string, nReduce int) *Master {
	m := Master{}

	// Your code here.
	m.inputfiles = files
	m.nReduce = nReduce
	m.ismap = true
	m.tasksfinished = 0
	i := 0
	for i < len(files) {
		m.taskstate = append(m.taskstate, AVAIL)
		i++
	}

	m.server()
	return &m
}
